package com.atguigu.gmall.realtime.dws.app;

import com.atguigu.gmall.realtime.common.base.BaseSQLApp;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.SQLUtil;
import com.atguigu.gmall.realtime.dws.function.keywordUDTF;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * ClassName: DwsTrafficSourceKeywordPageViewWindow
 * Description:
 *
 * @Create 2024/4/24 10:49
 */
public class DwsTrafficSourceKeywordPageViewWindow extends BaseSQLApp {

    public static void main(String[] args) {
        new DwsTrafficSourceKeywordPageViewWindow().start(10021, 4, "dws_traffic_source_keyword_page_view_window");
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        //TODO 注册自定义函数到表执行环境中
        tableEnv.createTemporaryFunction("ik_analyze", keywordUDTF.class);

        //TODO 从dwd的页面日志主题中读取数据，创建动态表，并指定事件时间以及WaterMark生成策略
        tableEnv.executeSql(
                "CREATE TABLE page_log\n" +
                        "(\n" +
                        "    common map<String,String>,\n" +
                        "    page map<String,String>,\n" +
                        "    ts bigint,\n" +
                        "    et as TO_TIMESTAMP_LTZ(ts,3),\n" +
                        "    WATERMARK FOR et AS et\n" +
                        ")" + SQLUtil.getKafkaDDL(Constant.TOPIC_DWD_TRAFFIC_PAGE, "dws_traffic_source_keyword_page_view_window")
        );

        //tableEnv.executeSql("select * from page_log").print();

        //TODO 过滤出搜索行为
        Table searchTable = tableEnv.sqlQuery(
                "select page['item'] fullword,\n" +
                        "       et\n" +
                        "from page_log\n" +
                        "where page['last_page_id'] = 'search'\n" +
                        "  and page['item_type'] = 'keyword'\n" +
                        "  and page['item'] is not null"
        );

        //searchTable.execute().print();

        tableEnv.createTemporaryView("search_table", searchTable);

        //TODO 对搜索内容进行分词，并和原表字段进行关联
        Table splitTable = tableEnv.sqlQuery(
                "SELECT keyword,\n" +
                        "       et\n" +
                        //","分开两个关联的表; Joins a table with the results of a table function; 一个表和一个表值函数结果集的Join
                        // t(keyword) -> 临时表名(表值函数返回字段);
                        "FROM search_table,\n" +
                        "     LATERAL TABLE(ik_analyze(fullword)) t(keyword);"
        );

        tableEnv.createTemporaryView("split_table", splitTable);

        //tableEnv.executeSql("select * from split_table").print();

        //TODO 分组、开窗、聚合
        Table resTable = tableEnv.sqlQuery(
                "SELECT \n" +
                        "   DATE_FORMAT(window_start, 'yyyy-MM-dd HH:mm:ss') stt, \n" +
                        "   DATE_FORMAT(window_end, 'yyyy-MM-dd HH:mm:ss') edt,\n" +
                        "   DATE_FORMAT(window_start, 'yyyy-MM-dd') cur_date, \n" +
                        "   keyword,\n" +
                        "   count(*) keyword_count\n" +
                        " FROM TABLE(\n" +
                        "    TUMBLE(TABLE split_table, DESCRIPTOR(et), INTERVAL '10' second))\n" +
                        "  GROUP BY keyword,window_start, window_end"
        );
        tableEnv.createTemporaryView("res_table", resTable);

        //tableEnv.executeSql("select * from res_table").print();


        //TODO 将聚合结果写到Doris。为什么一定要开检查点？
        tableEnv.executeSql(
                "create table dws_traffic_source_keyword_page_view_window\n" +
                        "(\n" +
                        "    stt string,\n" +
                        "    edt string,\n" +
                        "    cur_date string,\n" +
                        "    keyword string,\n" +
                        "    keyword_count bigint\n" +
                        ")" + SQLUtil.getDorisSinkDDL("dws_traffic_source_keyword_page_view_window")
        );

        resTable.executeInsert("dws_traffic_source_keyword_page_view_window");

    }
}
